package com.zdp.rocketmq.producer.tx;

import com.zdp.rocketmq.config.JmsConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

/**
 * @author sesshomaru
 * @date 2021/6/21
 * 事务消息生产者
 */
@Slf4j
@Component
public class TransactionProducerBuilder implements InitializingBean, DisposableBean {

    private String producerGroup = JmsConfig.TX_PRODUCER_GROUP;

    // 事务监听器(自定义)
    private TransactionListener transactionListener = new TransactionListenerImpl();

    private TransactionMQProducer transactionMQProducer = null;

    // 设置自定义线程池(用于接收回查)
    private ExecutorService executorService = new ThreadPoolExecutor(2,
            5, 100, TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName(producerGroup + "-check-thread");
            return thread;
        }
    });

    @Override
    public void afterPropertiesSet() throws Exception {
        transactionMQProducer = new TransactionMQProducer(producerGroup);
        transactionMQProducer.setNamesrvAddr(JmsConfig.NAME_SERVER);

        // 添加事务监听器(自定义)
        transactionMQProducer.setTransactionListener(transactionListener);
        // 添加自定义线程池(用于接收回查)
        transactionMQProducer.setExecutorService(executorService);

        transactionMQProducer.start();
    }

    @Override
    public void destroy() throws Exception {
        if (transactionMQProducer != null) {
            transactionMQProducer.shutdown();
        }
    }

    public TransactionMQProducer getProducer(){
        return this.transactionMQProducer;
    }
}


@Slf4j
class TransactionListenerImpl implements TransactionListener {

    // 指定本地事务并修改对应消息的状态
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        log.error("=============执行本地事务==========");
        String body = new String(msg.getBody());
        String key = msg.getKeys();
        String transactionId = msg.getTransactionId();
        log.info("transactionId="+transactionId+", key="+key+", body="+body);
        // 执行本地事务begin todo

        // 执行本地事务end todo

        int status = Integer.parseInt(arg.toString());


        if(status == 1){
            //提交事务 将 "半消息" 至为可见
            //二次确定消息，然后消费者可以消费
            // TODO 将transactionId和业务id绑定记录本地事务执行成功
            return LocalTransactionState.COMMIT_MESSAGE;
        }

        if(status == 2){
            //回滚消息,将以在broker中的 "半消息" 删除
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }

        if(status == 3){
            //broker端会进行回查消息
            return LocalTransactionState.UNKNOW;
        }

        return LocalTransactionState.ROLLBACK_MESSAGE;
    }

    // 当远端消息迟迟没有确认时，远端为主动回调这个方法校验本地事务是否执行成功，成功即可再次修改消息状态
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        log.error("=============回查校验本地事务==========");
        String body = new String(msg.getBody());
        String key = msg.getKeys();
        String transactionId = msg.getTransactionId();
        System.out.println("transactionId="+transactionId+", key="+key+", body="+body);

        //要么commit，要么是rollback
        //todo 根据transactionId检查本地事务是否成功 ok就commit 不ok就callback
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}
